/***************************************************************************
 *                                                                         *
 *                               NodeImpl.java                             *
 *                            -------------------                          *
 *   date                 : 16.08.2004                                     *
 *   copyright            : (C) 2004-2008 Distributed and                  *
 *                              Mobile Systems Group                       *
 *                              Lehrstuhl fuer Praktische Informatik       *
 *                              Universitaet Bamberg                       *
 *                              http://www.uni-bamberg.de/pi/              *
 *   email                : sven.kaffille@uni-bamberg.de                   *
 *   			    		karsten.loesing@uni-bamberg.de                 *
 *                                                                         *
 *                                                                         *
 ***************************************************************************/

/***************************************************************************
 *                                                                         *
 *   This program is free software; you can redistribute it and/or modify  *
 *   it under the terms of the GNU General Public License as published by  *
 *   the Free Software Foundation; either version 2 of the License, or     *
 *   (at your option) any later version.                                   *
 *                                                                         *
 *   A copy of the license can be found in the license.txt file supplied   *
 *   with this software or at: http://www.gnu.org/copyleft/gpl.html        *
 *                                                                         *
 ***************************************************************************/
package de.uniba.wiai.lspi.chord.service.impl;

import static de.uniba.wiai.lspi.util.logging.Logger.LogLevel.DEBUG;
import static de.uniba.wiai.lspi.util.logging.Logger.LogLevel.INFO;

import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Executor;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import de.uniba.wiai.lspi.chord.com.CommunicationException;
import de.uniba.wiai.lspi.chord.com.Endpoint;
import de.uniba.wiai.lspi.chord.com.Entry;
import de.uniba.wiai.lspi.chord.com.Node;
import de.uniba.wiai.lspi.chord.com.Proxy;
import de.uniba.wiai.lspi.chord.com.RefsAndEntries;
import de.uniba.wiai.lspi.chord.data.ID;
import de.uniba.wiai.lspi.chord.data.URL;
import de.uniba.wiai.lspi.chord.service.impl.Entries.AuxNode;
import de.uniba.wiai.lspi.util.logging.Logger;

/**
 * Implements all operations which can be invoked remotely by other nodes.
 * 
 * @author Karsten Loesing
 * @version 1.0.5
 */
public final class NodeImpl extends Node {

	/**
	 * Fusion
	 */
	public final static int MAX_FAILED_NODES = 1;
	public final static int MAX_PRIMARY_NODES = Integer
			.parseInt(System
					.getProperty("de.uniba.wiai.lspi.chord.service.impl.ChordImpl.StabilizeTask.start"));
	public boolean isFusionNode = false;
	public boolean isInFusionGroup = false;
	private FusedMap[] fusionRepository = null;
	private int fusedRefCount = 1;
	
	/**
	 * Endpoint for incoming communication.
	 */
	private Endpoint myEndpoint = null;

	/**
	 * Reference on local node.
	 */
	private ChordImpl impl;

	/**
	 * Object logger.
	 * The name of the logger is the name of this class with the nodeID appended. 
	 * The length of the nodeID depends on the number of bytes that are displayed 
	 * when the ID is shown in Hex-Representation. See documentation of {@link ID}. 
	 * E.g. de.uniba.wiai.lspi.chord.service.impl.NodeImpl.FF FF FF FF if the number 
	 * of displayed Bytes of an ID is 4. 
	 */
	private Logger logger;

	/**
	 * Routing table (including finger table, successor list, and predecessor
	 * reference)
	 */
	private References references;

	/**
	 * Repository for locally stored entries.
	 */
	private Entries entries;

	/**
	 * Executor that executes insertion and removal of entries on successors of
	 * this node.
	 */
	private Executor asyncExecutor;
	
	private Lock notifyLock; 

	/**
	 * Creates that part of the local node which answers remote requests by
	 * other nodes. Sole constructor, is invoked by ChordImpl only.
	 * 
	 * @param impl
	 *            Reference on ChordImpl instance which created this object.
	 * @param nodeID
	 *            This node's Chord ID.
	 * @param nodeURL
	 *            URL, on which this node accepts connections.
	 * @param references
	 *            Routing table of this node.
	 * @param entries
	 *            Repository for entries of this node.
	 * @throws IllegalArgumentException
	 *             If any of the parameter has value <code>null</code>.
	 */
	NodeImpl(ChordImpl impl, ID nodeID, URL nodeURL, References references,
			Entries entries) {

		if (impl == null || nodeID == null || nodeURL == null
				|| references == null || entries == null) {
			throw new IllegalArgumentException(
					"Parameters of the constructor may not have a null value!");
		}

		this.logger = Logger.getLogger(NodeImpl.class.getName() + "." + nodeID.toString());

		this.impl = impl;
		this.asyncExecutor = impl.getAsyncExecutor();
		this.nodeID = nodeID;
		this.nodeURL = nodeURL;
		this.references = references;
		this.entries = entries;
		this.notifyLock = new ReentrantLock(true); 
		
		// create endpoint for incoming connections
		this.myEndpoint = Endpoint.createEndpoint(this, nodeURL);
		this.myEndpoint.listen();
		
		/*Fusion*/
		fusionRepository = new FusedMap[fusedRefCount];
		for (int i = 0; i < fusedRefCount; ++i) {
			fusionRepository[i] = new FusedMap(MAX_PRIMARY_NODES, MAX_FAILED_NODES, i);
		}
	}

	/**
	 * Makes this endpoint accept entries by other nodes. Is invoked by
	 * ChordImpl only.
	 */
	final void acceptEntries() {
		this.myEndpoint.acceptEntries();
	}

	/**
	 * {@inheritDoc}
	 */
	@Override
	public final void disconnect() {
		/* Fusion */
		leavesFusionGroup();
		releaseFusionDuty();
		
		this.myEndpoint.disconnect();
	}

	/**
	 * {@inheritDoc}
	 */
	@Override
	public final Node findSuccessor(ID key) {
		return this.impl.findSuccessor(key);
	}

	/**
	 * {@inheritDoc}
	 */
	@Override
	public final List<Node> notify(Node potentialPredecessor) {
		/*
		 * Mutual exclusion between notify and notifyAndCopyEntries. 17.03.2008. sven.
		 */
		this.notifyLock.lock(); 
		try {
			// the result will contain the list of successors as well as the
			// predecessor of this node
			List<Node> result = new LinkedList<Node>();

			// add reference on predecessor as well as on successors to result
			if (this.references.getPredecessor() != null) {
				result.add(this.references.getPredecessor());
			} else {
				result.add(potentialPredecessor); 
			}
			result.addAll(this.references.getSuccessors());

//			 add potential predecessor to successor list and finger table and
			// set
			// it as predecessor if no better predecessor is available
			this.references.addReferenceAsPredecessor(potentialPredecessor);			
			return result;
		} finally {
			this.notifyLock.unlock(); 
		}
	}

	/**
	 * {@inheritDoc}
	 */
	@Override
	public final RefsAndEntries notifyAndCopyEntries(Node potentialPredecessor)
			throws CommunicationException {
		/*
		 * Mutual exclusion between notify and notifyAndCopyEntries. 17.03.2008. sven.
		 */
		this.notifyLock.lock(); 
		try {
			// copy all entries which lie between the local node ID and the ID
			// of
			// the potential predecessor, including those equal to potential
			// predecessor
			synchronized (this.entries) {
				Set<Entry> copiedEntries = this.entries.getEntriesInInterval(
						this.nodeID, potentialPredecessor.getNodeID());
			
				//Fusion: remove data in interval when a predecessor joined
				this.entries.removeEntriesInInterval(
						this.nodeID, potentialPredecessor.getNodeID(),
						this.references.getFusionNode(), this.nodeID);
	
				return new RefsAndEntries(this.notify(potentialPredecessor),
						copiedEntries);
			}
		} finally {
			this.notifyLock.unlock(); 
		}
	}

	/**
	 * {@inheritDoc}
	 */
	@Override
	public final void ping() throws CommunicationException {
		//if (this.myEndpoint.getState() == Integer.MAX_VALUE) throw new CommunicationException();
		// do nothing---returning of method is proof of live
		return;
	}

	/**
	 * {@inheritDoc}
	 */
	@Override
	public final void insertEntry(Entry toInsert) {//throws CommunicationException {
		if (this.logger.isEnabledFor(DEBUG)) {
			this.logger.debug("Inserting entry with id " + toInsert.getId()
					+ " at node " + this.nodeID);
		}

		// Possible, but rare situation: a new node has joined which now is
		// responsible for the id!
		/*
		if ((this.references.getPredecessor() == null)
				|| !toInsert.getId().isInInterval(
						this.references.getPredecessor().getNodeID(),
						this.nodeID)) {
			this.references.getPredecessor().insertEntry(toInsert); 
			return;
		}
		//*/
		
		// Fusion: insert data into fusion server
		// entry set for updating fusion server
		Set<Entry> oldEntrySet = null;
		Set<Entry> newEntrySet = null;
		ID entryId = toInsert.getId();

		// get the entry set before adding entryToAdd
		synchronized (this.entries) {
			if (this.references.getFusionNode() != null) {
				if (this.entries.getEntries().containsKey(entryId)) {
					oldEntrySet = new HashSet<Entry>(this.entries.getEntries().get(entryId).value);
				}
			}
			
			// add entry to local repository
			this.entries.add(toInsert);
	
			// get the updated entry set after inserting data into fusion server
			if (this.references.getFusionNode() != null) {
				newEntrySet = this.entries.getEntries().get(toInsert.getId()).value;
				//System.err.println(this.getNodeURL() + " is inserting data into fusion server " + this.references.getFusionNode().getNodeURL());
				try {
					this.references.getFusionNode().fusionAddEntry(entryId, newEntrySet, oldEntrySet, this.getNodeID());
				} catch (CommunicationException e) {
					e.printStackTrace();
				}
			}
		}

		// Fusion: Remove replicate.
		/*
		// create set containing this entry for insertion of replicates at all
		// nodes in successor list
		Set<Entry> newEntries = new HashSet<Entry>();
		newEntries.add(toInsert);

		// invoke insertReplicates method on all nodes in successor list
		final Set<Entry> mustBeFinal = new HashSet<Entry>(newEntries);
		for (final Node successor : this.references.getSuccessors()) {
			this.asyncExecutor.execute(new Runnable() {
				public void run() {
					try {
						successor.insertReplicas(mustBeFinal);
					} catch (CommunicationException e) {
						// do nothing
					}
				}
			});
		}
		*/
	}

	/**
	 * {@inheritDoc}
	 */
	@Override
	public final void insertReplicas(Set<Entry> replicatesToInsert) {
		// Fusion: no more replicate!
		//this.entries.addAll(replicatesToInsert);
	}

	/**
	 * {@inheritDoc}
	 */
	@Override
	public final void removeEntry(Entry entryToRemove)
			throws CommunicationException {

		if (this.logger.isEnabledFor(DEBUG)) {
			this.logger.debug("Removing entry with id " + entryToRemove.getId()
					+ " at node " + this.nodeID);
		}

		// Possible, but rare situation: a new node has joined which now is
		// responsible for the id!
		if (this.references.getPredecessor() != null
				&& !entryToRemove.getId().isInInterval(
						this.references.getPredecessor().getNodeID(),
						this.nodeID)) {
			this.references.getPredecessor().removeEntry(entryToRemove);
			return;
		}

		ID entryId = entryToRemove.getId();
		synchronized (this.entries) {
			// make sure the entryToMove exists in local repository
			if (this.entries.getEntries().containsKey(entryId)) {
				// entry set for updating fusion server
				Set<Entry> oldEntrySet = null;
				Set<Entry> newEntrySet = null;
				Set<Entry> finalEntrySet = null;
				
				// get the entry set before removing entryToMove
				if (this.references.getFusionNode() != null) {
					oldEntrySet = new HashSet<Entry>(this.entries.getEntries().get(entryId).value);
				}
			
				// remove entry from repository
				finalEntrySet = this.entries.remove(entryToRemove);
				
				// update fusion server if current node has a fusion group
				if (this.references.getFusionNode() != null) {
					// check if we need to update the entry set or remove entry set from fusion server
					if (this.entries.getEntries().containsKey(entryId)) {
						// update fusion server
						newEntrySet = this.entries.getEntries().get(entryId).value;
						this.references.getFusionNode().fusionAddEntry(entryId, oldEntrySet, newEntrySet, this.getNodeID());
					} else {
						//remove entry set from fusion server
						//finalEntrySet = this.entries.getFinalPrimary();
						if (finalEntrySet == null) finalEntrySet = oldEntrySet;
						this.references.getFusionNode().fusionRemoveEntry(entryId, oldEntrySet, finalEntrySet, this.getNodeID());
					}
				}
			}
		}

		// Fusion: remove replicate
		/* 
		// create set containing this entry for removal of replicates at all
		// nodes in successor list
		final Set<Entry> entriesToRemove = new HashSet<Entry>();
		entriesToRemove.add(entryToRemove);

		// invoke removeReplicates method on all nodes in successor list
		List<Node> successors = this.references.getSuccessors();
		final ID id = this.nodeID;
		for (final Node successor : successors) {
			this.asyncExecutor.execute(new Runnable() {
				public void run() {
					try {
						// remove only replica of removed entry
						successor.removeReplicas(id, entriesToRemove);
					} catch (CommunicationException e) {
						// do nothing for the moment
					}
				}
			});
		}
		//*/
	}

	/**
	 * {@inheritDoc}
	 */
	@Override
	public final void removeReplicas(ID sendingNodeID,
			Set<Entry> replicasToRemove) {
		//Fusion: remove replicate
		/*
		if (replicasToRemove.size() == 0) {
			// remove all replicas in interval
			boolean debug = this.logger.isEnabledFor(DEBUG);
			if (debug) {
				this.logger.debug("Removing replicas. Current no. of entries: "
						+ this.entries.getNumberOfStoredEntries());
			}
			/*
			 * Determine entries to remove. These entries are located between
			 * the id of the local peer and the argument sendingNodeID
			 */
		/*	Set<Entry> allReplicasToRemove = this.entries.getEntriesInInterval(
					this.nodeID, sendingNodeID);
			if (debug) {
				this.logger.debug("Replicas to remove " + allReplicasToRemove);
				this.logger.debug("Size of replicas to remove "
						+ allReplicasToRemove.size());
			}

			/*
			 * Remove entries
			 */
		/*	this.entries.removeAll(allReplicasToRemove);

			if (debug) {
				this.logger
						.debug("Removed replicas??? Current no. of entries: "
								+ this.entries.getNumberOfStoredEntries());
			}
		} else {
			// remove only replicas of given entry
			this.entries.removeAll(replicasToRemove);
		}
		//*/
	}

	/**
	 * {@inheritDoc}
	 */
	@Override
	public final Set<Entry> retrieveEntries(ID id)
			throws CommunicationException {

		// Possible, but rare situation: a new node has joined which now is
		// responsible for the id!
		if (this.references.getPredecessor() != null
				&& !id.isInInterval(this.references.getPredecessor()
						.getNodeID(), this.nodeID)) {
			this.logger.fatal("The rare situation has occured at time "
					+ System.currentTimeMillis() + ", id to look up=" + id
					+ ", id of local node=" + this.nodeID
					+ ", id of predecessor="
					+ this.references.getPredecessor().getNodeID());
			return this.references.getPredecessor().retrieveEntries(id);
		}

		// return entries from local repository
		// for this purpose create a copy of the Set in order to allow the
		// thread retrieving the entries to modify the Set without modifying the
		// internal Set of entries. sven
		return this.entries.getEntries(id);
	}

	/**
	 * {@inheritDoc}
	 */
	@Override
	final public void leavesNetwork(Node predecessor) {
		if (this.logger.isEnabledFor(INFO)) {
			this.logger.info("Leaves network invoked; " + this.nodeID
					+ ". Updating references.");
			this.logger.info("New predecessor " + predecessor.getNodeID());
		}
		if (this.logger.isEnabledFor(DEBUG)) {
			this.logger.debug("References before update: "
					+ this.references.toString());
		}
		this.references.removeReference(this.references.getPredecessor());
		if (this.logger.isEnabledFor(DEBUG)) {
			this.logger.debug("References after update: "
					+ this.references.toString());
		}
	}

	/**
	 * 
	 * @return
	 */
	final Executor getAsyncExecutor() {
		return this.asyncExecutor;
	}
	

	/**
	 *  Fusion
	 */
	public Node getPredecessor() {
		return this.impl.references.getPredecessor();
	}

	@Override
	public boolean isFusionNode() {
		return this.isFusionNode;
	}
	
	/** Primary Node's method **/
	@Override
	public void joinFusionGroup() {
		if (this.references.getFusionNode() != null) {
			System.err.println(this.getNodeURL() + " is doing weird things.");
		}
		Node FusionNode = findFusionGroup();
		if (FusionNode == null) {
			FusionNode = requestForCreatingFusionGroup();
		}

		if (FusionNode != null) {
			this.references.setFusionNode(FusionNode);
			System.err.println(this.getNodeURL() + " joined a Fusion group: " + FusionNode.getNodeURL());
			try {
				synchronized (this.entries) {
					FusionNode.initializeFusionData(this.entries, this.getNodeID());
				}
			} catch (CommunicationException e) {
				e.printStackTrace();
			}
		}
		// System.err.println(this.getNodeURL() + " failed to join a Fusion group.");
	}

	private void leavesFusionGroup() {
		try {
			if (this.references.getFusionNode() != null) {
				this.references.getFusionNode().requestLeaveFusionGroup(this.getNodeID());
				// TODO transfer the data to erase to fusion server
			}
		} catch (CommunicationException e) {
			e.printStackTrace();
		}
	}
	
	// return null if cannot find one
	private Node findFusionGroup() {
		Node FusionNode = getPredecessor();
		
		while (FusionNode != null) {
			if (FusionNode.equals(this)) {
				FusionNode = null;
				break;
			}
			
			// failed to join a group, then find next one
			try {
				if (!FusionNode.isFusionNode() || !FusionNode.requestJoinFusionGroup(this.getNodeURL())) {
					FusionNode = FusionNode.getPredecessor();
				} else {
					break;
				}
			} catch (CommunicationException e) {
				FusionNode = null;
				break;
			}
		}
		
		if (FusionNode != null) {
			try {
				FusionNode = Proxy.createConnection(this.getNodeURL(), FusionNode.getNodeURL());
			} catch (CommunicationException e) {
				e.printStackTrace();
			}
		}
		
		return FusionNode;
	}
	
	// request another primary node to start a fusion group
	private Node requestForCreatingFusionGroup() {
		Node FusionNode = getPredecessor();
		
		while (FusionNode != null) {
			if (FusionNode.equals(this)) {
				FusionNode = null;
				break;
			}
			
			try {
				if (FusionNode.isFusionNode()) {
					if (FusionNode.requestJoinFusionGroup(this.getNodeURL())) {
						break;
					} else {
						FusionNode = FusionNode.getPredecessor();
					}
				} else {
					if (FusionNode.requestCreateFusionGroup(this.getNodeURL())) {
						break;
					} else {
						FusionNode = FusionNode.getPredecessor();
					}
				}
			} catch (CommunicationException e) {
				FusionNode = null;
				break;
			}
		}
		
		if (FusionNode != null) {
			try {
				FusionNode = Proxy.createConnection(this.getNodeURL(), FusionNode.getNodeURL());
			} catch (CommunicationException e) {
				e.printStackTrace();
			}
		}
		
		return FusionNode;
	}
	
	// update primaries' fusion node pointer
	@Override
	public void updateFusionNode(Node n) {
		this.references.setFusionNode(n);
	}

	// return true if node n successfully joined the group
	@Override
	public synchronized boolean requestCreateFusionGroup(URL primaryUrl) throws CommunicationException {
		if (this.isFusionNode) {
			return false;
		} else {
			this.isFusionNode = true;
			this.references.initialPaimaryList();
			this.addPrimary(Proxy.createConnection(this.getNodeURL(), primaryUrl));
			
			return true;
		}
	}
	
	public synchronized boolean receiveFusedData(List<Node> primaries) {
		if (this.isFusionNode) {
			System.err.println(this.getNodeURL() + " is a fusion server already.");
			return false;
		} else {
			this.isFusionNode = true;
			this.references.initialPaimaryList();
			//this.references.getPrimaryNodes().addAll(primaries);
			for (int i = 0; i < this.MAX_PRIMARY_NODES; ++i) {
				this.references.getPrimaryNodes().set(i, primaries.get(i));
			}
			
			// TODO receive fused data here
			
			return true;
		}
	}
	
	public Entries transferEntriesForRecover() {
		synchronized (this.entries) {
			return this.entries;
		}
	}
	
	
	
	
	
	
	/** Fusion Node's method 
	 * @throws CommunicationException **/
	@Override
	// return true if node n successfully joined the group
	public synchronized boolean requestJoinFusionGroup(URL primaryUrl) throws CommunicationException {
		if (this.references.numberOfPrimaries() < MAX_PRIMARY_NODES) {
			addPrimary(Proxy.createConnection(this.getNodeURL(), primaryUrl));
			return true;
		} else { 
			return false;
		}
	}
	
	@Override
	public synchronized void requestLeaveFusionGroup(ID nodeId) throws CommunicationException {
		if (this.isFusionNode) {
			removePrimary(nodeId);
		}
	}
	
	@Override
	public synchronized void fusionAddEntry(ID entryId, Set<Entry> newEntries, Set<Entry> oldEntries, ID primaryNodeId) {
		//System.err.println(this.getNodeURL() + "is adding data from " + primaryNodeId + "into fusion repository.");
		if (entryId == null || primaryNodeId == null || newEntries == null) {
			System.err.println("Error occurred when adding entry into a fused map: null pointer");
		}
		int primaryId = this.references.getPrimaryIndex(primaryNodeId);
		if (primaryId >= 0) {
			fusionRepository[0].put(entryId, newEntries, oldEntries, primaryId);
		} else {
			System.err.println("Error occurred when adding entry into a fused map.");
		}
	}

	@Override
	public synchronized void fusionRemoveEntry(ID entryId, Set<Entry> oldEntrySet, Set<Entry> finalEntrySet, ID primaryNodeId) {
		//System.err.println(this.getNodeURL() + "is removing data from " + primaryNodeId + "from fusion repository.");
		if (entryId == null || oldEntrySet == null || finalEntrySet == null || primaryNodeId == null) {
			System.err.println("Error occurred when removing entry from a fused map: null pointer.");
		}
		int primaryId = this.references.getPrimaryIndex(primaryNodeId);
		if (primaryId >= 0 && primaryId < this.MAX_PRIMARY_NODES) {
			fusionRepository[0].remove(entryId, oldEntrySet, finalEntrySet, primaryId);
		} else {
			System.err.println("Error occurred when removing entry from a fused map: Primay ID is out of range.");
		}
	}
	
	@Override
	public void recoverPrimaryData(int crashedPrimaryIndex) throws CommunicationException {
		// get all healthy entries
		Entries[] entriesList = new Entries[MAX_PRIMARY_NODES];
		for (int i = 0; i < MAX_PRIMARY_NODES; ++i) {
			entriesList[i] = null;
		}
		// get healthy Entries from primaries
		for (int primaryIndex = 0; primaryIndex < MAX_PRIMARY_NODES; ++primaryIndex) {
			if (primaryIndex != crashedPrimaryIndex) {
				Node parimaryNode = this.references.getPrimary(primaryIndex);
				if (parimaryNode != null) {
					entriesList[primaryIndex] = new Entries(parimaryNode.transferEntriesForRecover());
				}
			}
		}
		
		// inserting recovered data into Chord network
		Set<Entry> recoveredData = fusionRepository[0].recover(crashedPrimaryIndex, entriesList);
		for (Entry recoveredEntry : recoveredData) {
			this.impl.insert(recoveredEntry);
		}
	}
	
	public synchronized void initializeFusionData(Entries entries, ID primaryNodeId) throws CommunicationException {
		int primaryIndex = this.references.getPrimaryIndex(primaryNodeId);
		if (primaryIndex >= 0 && primaryIndex < this.MAX_PRIMARY_NODES) {
			for (AuxNode auxNode : entries.auxList) {
				Set<Entry> newEntries = auxNode.priNode.value;
				Set<Entry> oldEntries = null;
				ID entryId = newEntries.iterator().next().getId();
				fusionRepository[0].put(entryId, newEntries, oldEntries, primaryIndex);
			}
		} else {
			System.err.println("Error occurred when initializing a whole new entries into a fused map.");
		}
	}

	private synchronized void addPrimary(Node proxy) {
		if (!this.references.addPrimary(proxy)) {
			System.err.println(proxy.getNodeURL() + " is trying to join this Fusion group (" + this.getNodeURL()+ ") failed!");
			return;
		}
	}
	
	public synchronized void removePrimary(ID primaryNodeId) {
		this.fusionRepository[0].auxStructure.set(this.references.getPrimaryIndex(primaryNodeId), new HashMap<ID, FusedAuxNode>());
		this.references.removePrimary(primaryNodeId);
	}
	
	private synchronized void releaseFusionDuty() {
		if (this.isFusionNode) {
			Node FusionNodeCandidate = null;
			int counter = 0;
			// looking for replacement 
			while (FusionNodeCandidate == null && counter < 5) {
				try {
					FusionNodeCandidate = findFusionNodeCandidate();
				} catch (CommunicationException e1) {
					break;
				}
				synchronized (this) {
					try {
						wait(500);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
				++counter;
			}
			if (FusionNodeCandidate != null) {
				// tell primaries the new candidate
				try { // in case of a primary node dies
					for(Node n : this.references.getPrimaryNodes()) {
						if (n != null) {
							n.updateFusionNode(FusionNodeCandidate);
						}
					}
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}
	}
	
	// ask a primary node to take its place and become a fusion node
	private Node findFusionNodeCandidate() throws CommunicationException {
		Node FusionNode = getPredecessor();
		
		while (FusionNode != null) {
			if (FusionNode.equals(this)) {
				FusionNode = null;
				break;
			}
			
			if (FusionNode.isFusionNode() || this.references.containPrimary(FusionNode.getNodeID())
				|| !FusionNode.receiveFusedData(this.references.getPrimaryNodes())) { // transfer fused data to the new fusion node
				FusionNode = FusionNode.getPredecessor();
			} else {
				System.err.println("Transfering fused data to new fusion node.");
				break;
			}
		}
		
		return FusionNode;
	}
	
	// Misc
	public void addAllEntries(Set<Entry> copyOfEntries) {
		if (copyOfEntries == null) {
			NullPointerException e = new NullPointerException(
					"Set of entries to be added to the local hash table may "
							+ "not be null!");
			this.logger.error("Null pointer", e);
			throw e;
		}

		for (Entry nextEntry : copyOfEntries) {
			this.insertEntry(nextEntry);
		}
	}
	
	public String printFusedData() {
		return this.fusionRepository[0].toString();
	}
}
